package com.eric.study.group;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.*;
import com.eric.study.device.Device;
import com.eric.study.manager.DeviceManager;
import com.eric.study.message.Message;

import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * <p>
 * DeviceGroupQuery
 *
 * <p>
 * change history:
 *
 * <pre>
 * date         defect       person          comments
 * --------------------------------------------------------------------------------
 * 2023/8/19    ********     Eric    create file.
 * </pre>
 *
 * @author Eric
 * @date 2023/8/19 17:02
 * @since 1.0.0
 */
public class DeviceGroupQuery extends AbstractBehavior<DeviceGroupQuery.GroupQueryMessage> {

    /**
     * 请求编号
     */
    private final long requestId;

    /**
     * 原始请求查询者
     */
    private final ActorRef<DeviceManager.RespondAllTemperatures> requester;

    /**
     * 已经响应的设备集合。key - 设备编号，value - 设备温度响应结果
     */
    private final Map<String, DeviceManager.TemperatureReading> repliesSoFar;

    /**
     * 所有的设备消息回复等待队列。向设备发送了查询温度命令后，就加入此队列中。
     */
    private final Set<String> stillWaiting;


    private DeviceGroupQuery(ActorContext<GroupQueryMessage> context,
                             Map<String, ActorRef<Device.DeviceMessage>> deviceActor,
                             long requestId,
                             ActorRef<DeviceManager.RespondAllTemperatures> requester,
                             Duration timeout,
                             TimerScheduler<GroupQueryMessage> timers) {
        super(context);
        this.requestId = requestId;
        this.requester = requester;

        // 将响应结果集合设置为空 Map
        repliesSoFar = new HashMap<>();

        // 启动一个定时器，当时间超过给定的限制时，给自己发送一个超时命令
        timers.startSingleTimer(CollectionTimeout.INSTANCE, timeout);

        // 将设备 Actor 的响应消息做一个适配，转化为当前 Actor 可以接收的协议
        ActorRef<Device.RespondTemperature> respondedAdapter = context.messageAdapter(Device.RespondTemperature.class, WrappedRespondTemperature::new);

        for (Map.Entry<String, ActorRef<Device.DeviceMessage>> entry : deviceActor.entrySet()) {
            // 观察每个设备的状态，如果 Actor 发生异常停止了，则向自己发送一个命令
            context.watchWith(entry.getValue(), new DeviceTerminated(entry.getKey()));
            // 向每一个设备发送获取温度的命令
            entry.getValue().tell(new Device.ReadTemperature(0L, respondedAdapter));
        }
        // 将所有要查询的设备 Actor 放入等待队列中
        stillWaiting = new HashSet<>(deviceActor.keySet());
    }

    public static Behavior<GroupQueryMessage> create(Map<String, ActorRef<Device.DeviceMessage>> deviceActor,
                                                     long requestId,
                                                     ActorRef<DeviceManager.RespondAllTemperatures> requester,
                                                     Duration timeout) {
        return Behaviors.setup(context ->
                Behaviors.withTimers(timers -> new DeviceGroupQuery(context, deviceActor, requestId, requester, timeout, timers)));
    }

    @Override
    public Receive<GroupQueryMessage> createReceive() {
        return newReceiveBuilder()
                // 处理设备响应成功的消息
                .onMessage(WrappedRespondTemperature.class, this::onRespondTemperature)
                // 处理设备因异常关闭的消息
                .onMessage(DeviceTerminated.class, this::onDeviceTerminated)
                // 处理查询超时的消息
                .onMessage(CollectionTimeout.class, this::onCollectionTimout)
                .build();
    }

    /**
     * 检查等待队列中是否有剩余的设备编号。如果有，返回当前对象，并等待更多消息回复；否则，代码所有查询的设备都有返回结果，将
     * 返回结果发送给原始请求者，并停止当前的查询 Actor。
     *
     * @author Eric
     */
    private Behavior<GroupQueryMessage> respondWhenAllCollected() {
        if (stillWaiting.isEmpty()) {
            requester.tell(new DeviceManager.RespondAllTemperatures(requestId, repliesSoFar));
            return Behaviors.stopped();
        } else {
            return this;
        }
    }


    private Behavior<GroupQueryMessage> onRespondTemperature(WrappedRespondTemperature msg) {
        // 转化从设备Actor中获取的结果值
        DeviceManager.TemperatureReading reading = msg.response()
                .value()
                .map(v -> (DeviceManager.TemperatureReading) new DeviceManager.Temperature(v))
                .orElse(DeviceManager.TemperatureNotAvailable.INSTANCE);

        // 获取返回结果的设备编号
        String deviceId = msg.response.deviceId();
        // 将转化结果添加到响应消息集合中
        repliesSoFar.put(deviceId, reading);
        // 将设备从等待回复消息队列中移除
        stillWaiting.remove(deviceId);

        return respondWhenAllCollected();
    }

    /**
     * 处理设备异常停止的命令
     *
     * @param msg 待处理的命令
     * @author Eric
     */
    private Behavior<GroupQueryMessage> onDeviceTerminated(DeviceTerminated msg) {
        if (stillWaiting.contains(msg.deviceId)) {
            repliesSoFar.put(msg.deviceId, DeviceManager.DeviceNotAvailable.INSTANCE);
            stillWaiting.remove(msg.deviceId);
        }

        return respondWhenAllCollected();
    }

    private Behavior<GroupQueryMessage> onCollectionTimout(CollectionTimeout msg) {
        for (String deviceId : stillWaiting) {
            // 将剩余未回复的 Actor 的响应设置为超时
            repliesSoFar.put(deviceId, DeviceManager.DeviceTimedOut.INSTANCE);
        }
        // 清除所有回复等待队列
        stillWaiting.clear();

        return respondWhenAllCollected();
    }


    public interface GroupQueryMessage extends Message {
    }


    /**
     * 查询超时的消息
     */
    private enum CollectionTimeout implements GroupQueryMessage {
        INSTANCE
    }

    /**
     * 设备关闭命令
     *
     * @param deviceId 设备编号
     */
    private record DeviceTerminated(String deviceId) implements GroupQueryMessage {
    }

    /**
     * 消息适配器
     *
     * @param response 查询设备温度返回的结果响应信息
     */
    public record WrappedRespondTemperature(Device.RespondTemperature response) implements GroupQueryMessage {
    }

}
